package com.freez.spark.core.demo;

import com.clearspring.analytics.util.Lists;
import com.freez.spark.tool.TupleUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: SecondSort.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:25
 * Description: 电影二次排序
 */
public class SecondSort {
    public static void main(String[] args) {
        String appName1 = "SparkDemo-secondSort";
        String master = "spark://slaveNode2:7077";
        String path = "./src/main/resources/data/core/demo/secondarySort.txt";
        secondSort(appName1, master, path);
    }
    /**
     * 二次排序
     * 战狼2,2017,56.39
     * 年份:2018 票房(亿元):36.22 片名:红海行动
     *
     * @param appName Application的名字
     * @param master  设置Spark程序的运行模式
     * @param path    源文件路径
     */
    public static void secondSort(String appName, String master, String path) {
        SparkConf conf = new SparkConf();
        conf.setAppName(appName).setMaster(master);
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> txtRDD = jsc.textFile(path);
        JavaPairRDD<Integer, Tuple2<String, Double>> pairRDD = txtRDD.mapToPair(s -> {
            String[] split = s.split(",");
            String movieName = split[0];
            Integer date = Integer.parseInt(split[1]);
            Double boxOffice = Double.parseDouble(split[2]);
            return TupleUtils.T2(date, TupleUtils.T2(movieName, boxOffice));
        });
        JavaPairRDD<Integer, Iterable<Tuple2<String, Double>>> firstSortRDD = pairRDD.groupByKey().sortByKey(false);
        JavaPairRDD<Integer, List<Tuple2<String, Double>>> secondSortRDD = firstSortRDD.mapValues(s -> {
//            List<Tuple2<String, Double>> list = new ArrayList<>((Collection<? extends Tuple2<String, Double>>) s);
            List<Tuple2<String, Double>> list = Lists.newArrayList(s);
//            Collections.sort(list, new TupleComparator());
            Collections.sort(list, new Comparator<Tuple2<String, Double>>() {
                @Override
                public int compare(Tuple2<String, Double> o1, Tuple2<String, Double> o2) {
                    if (o1._2 > o2._2) {
                        return -1;
                    }
                    return 1;
                }
            });
            return list;
        });
        JavaPairRDD<Integer, Tuple2<String, Double>> resultRDD = secondSortRDD.flatMapToPair(s -> {
            Integer date = s._1;
            List<Tuple2<Integer, Tuple2<String, Double>>> resultList = new ArrayList<>();
            List<Tuple2<String, Double>> list = s._2;
            if (list.size() < 2) {
                resultList.add(TupleUtils.T2(date, list.get(0)));
                return resultList.iterator();
            }
            for (Tuple2<String, Double> item : list) {
                resultList.add(TupleUtils.T2(date, item));
            }
            return resultList.iterator();
        });
        //打印结果
        for (Tuple2<Integer,Tuple2<String,Double>> s : resultRDD.collect()){
            System.out.println("年份：" + s._1 + " 票房：" + s._2._2 + "(亿元) 片名：" + s._2._1);
        }
    }
}